package com.sbp.message.send_message.impl;

import com.alibaba.fastjson.JSON;
import com.dap.utils.AsyncQueue;
import com.sbp.message.api.entity.Message;
import com.sbp.message.api.entity.MessageType;
import com.sbp.message.send_message.MessageSender;
import org.apache.activemq.command.ActiveMQTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import javax.jms.JMSException;
import javax.jms.Session;

@Service("messageSender")
public class MessageSenderImpl implements MessageSender, AsyncQueue.AsyncQueueConsumer {
    private static final Logger logger = LoggerFactory.getLogger(MessageSenderImpl.class);

    private AsyncQueue asyncQueue = new AsyncQueue();

    @Resource(name = "queueJmsTemplate")
    private JmsTemplate queueJmsTemplate;

    @Resource(name = "topicJmsTemplate")
    private JmsTemplate topicJmsTemplate;

    @Override
    public void sendMessage(final Message message) {
        if (validateMessage(message)) {
            asyncQueue.enqueueMessage(this, message, message.getCreateTime().getTime() + 1000);
        } else {
            logger.warn("invalid message={}", JSON.toJSONString(message));
        }
    }

    private void realSendMessage(final Message message) {
        if (message.getMessageType() == MessageType.MESSAGE_QUEUE.getValue()) {
            queueJmsTemplate.send(message.getDestination(), new MessageCreator() {
                public javax.jms.Message createMessage(Session session) throws JMSException {
                    return session.createTextMessage(JSON.toJSONString(message));
                }
            });
        } else {
            topicJmsTemplate.send(new ActiveMQTopic(message.getDestination()), new MessageCreator() {
                public javax.jms.Message createMessage(Session session) throws JMSException {
                    return session.createTextMessage(JSON.toJSONString(message));
                }
            });
        }
    }

    private static boolean validateMessage(Message message) {
        if (null == message) {
            return false;
        } else {
            if (StringUtils.isEmpty(message.getDestination())) {
                return false;
            }
            if (message.getMessageType() != MessageType.MESSAGE_TOPIC.getValue()) {
                message.setMessageType(MessageType.MESSAGE_QUEUE.getValue());
            }
        }

        return true;
    }

    @Override
    public void close() {
        asyncQueue.close();
    }

    @Override
    public void consumeAsyncQueueContent(Object o) {
        realSendMessage((Message) o);
    }
}
